package com.bieber.computer.splitter;

import com.bieber.common.Constants;
import com.bieber.common.model.CacheValue;
import com.bieber.common.model.ConsumerPerson;
import com.bieber.computer.ComputerConfig;
import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCache;
import org.apache.ignite.cache.CacheAtomicityMode;
import org.apache.ignite.cache.CacheMode;
import org.apache.ignite.cache.CachePeekMode;
import org.apache.ignite.configuration.CacheConfiguration;

import javax.xml.crypto.Data;

/**
 * Created by bieber on 2015/9/17.
 */
public abstract class DataSplitter<T extends CacheValue> implements Runnable {

    protected volatile int index=0;
    
    protected volatile ComputerConfig config;
    
    protected volatile IgniteCache<Integer, CacheValue<T>> dataCache;
    
    protected Ignite ignite;
    
    protected String cacheName;
    
    protected volatile boolean isFinished;
    
    public DataSplitter(ComputerConfig config,Ignite ignite){
        this.config = config;
        this.ignite = ignite;
    }
    public int getCacheSize(){
        return dataCache.size(CachePeekMode.PRIMARY);
    }
    public void cleanCache(){
        dataCache.removeAll();
    }
    
    public boolean isFinished(){
        return isFinished;
    }
    
    @Override
    public void run() {
        split();
    }

    protected abstract void split();
    

    public String getCacheName(){
        return cacheName;
    }
    
    protected void initCache(String cacheName,Class<T> type){
        this.cacheName = cacheName;
        CacheConfiguration<Integer,CacheValue<T>> personCacheConfiguration = new CacheConfiguration<Integer, CacheValue<T>>(cacheName);
        personCacheConfiguration.setCacheMode(CacheMode.PARTITIONED);
        personCacheConfiguration.setBackups(1);
        personCacheConfiguration.setAtomicityMode(CacheAtomicityMode.TRANSACTIONAL);
        personCacheConfiguration.setIndexedTypes(Integer.class, type);
        dataCache = ignite.getOrCreateCache(personCacheConfiguration);
    }
}
